package com._58city.spark.app.mr;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import scala.Tuple2;

import com._58city.spark.app.DaoUtil;
import com._58city.spark.app.SparkUtil;
import com._58city.spark.app.dao.TrackDto;
import com._58city.spark.app.dao.TrackValue;
import com._58city.spark.app.ext.JsonUtil;
import com._58city.spark.app.redis.Constant;
import com._58city.spark.app.redis.RedisUtil;
import com._58city.spark.app.redis.TrackModel;
import com._58city.spark.app.redis.TrackSearchWord;
import com._58city.spark.app.redis.TransferDimensionUtil;

/**
 * @author lihao
 * 流量的Spark运算类
 */
public class MrKafkaTrack implements MrKafkaInterface<String, TrackValue>{

	private static final long serialVersionUID = 5276176426349314034L;
	private Logger log = LoggerFactory.getLogger(MrKafkaTrack.class);
    
    private static final ExecutorService threadPool = Executors.newCachedThreadPool();
	
	private String[] keys;
	
	private  final int batchInterval;
	
    private final int platform_type; //PC(-1), M(0), APP(2)
    
    //计算实时总UV，将cookie值存入redis集合,由dervice服务每10秒取一次集合大小即当前UV
    private final String cookie_redis_key;
	
	public MrKafkaTrack(String[] keys, int batchInterval, int platform_type) {
		super();
		this.keys = keys;
		this.batchInterval = batchInterval;
		this.platform_type = platform_type;
		switch (platform_type) {
		case -1:
			cookie_redis_key = Constant.COOKIE_REDIS_QUEUE_KEY_PC;
			break;
		case 0:
			cookie_redis_key = Constant.COOKIE_REDIS_QUEUE_KEY_M;
			break;
		case 2:
			cookie_redis_key = Constant.COOKIE_REDIS_QUEUE_KEY_APP;
			break;
		default:
			cookie_redis_key = "cookie_queue_error";
			break;
		}	
	}
	
	/**
	 * 从日志中找到搜索词,站内
	 * @param url
	 * @return
	 */
	public static String get_search_word(String url){
		String  keyWordReStr = "(?:/key_|[&?]key=|/jh_)([^?^/^&]+)";
		Pattern keyWordRePatten = Pattern.compile(keyWordReStr) ;
		Matcher matc = keyWordRePatten.matcher(url);
		if (!matc.find()) {
			return "";
		}
		return matc.group(1);
//		return YURLDecode.searchWordDecode(matc.group(1));
	}
	
	//【站外搜索词】没有用到
    @SuppressWarnings("unused")
	private static String getWord(String src)
    {   
        String word = null;
        if(StringUtils.isNotBlank(src))
        {
            String pattern = "(\\?|&)(wd=|word=|q=|query=|w=|kw=|keyword=|p=)";
            Pattern p_a = Pattern.compile(pattern, 2);
            Matcher m_a = p_a.matcher(src);
            if(m_a.find())
            {
                String a = m_a.group();
                if(a.contains("?"))
                    a = (new StringBuilder()).append("\\").append(a).toString();
                String srcs[] = src.split(a, -1);
                if(srcs.length > 1)
                {
                    int index = srcs[1].indexOf("&");
                    int tmpIndex = index != -1 ? index : srcs[1].length();
                    word = srcs[1].substring(0, tmpIndex);
                } else
                {
                    word = srcs[0];
                }
            }
        }
        return word;
    }
    
	/**
	 * 将日志转化为一个流量的Bean对象
	 * @param content
	 * @return
	 */
	private TrackDto toDto(String content){
		if(StringUtils.isNotBlank(content)){
			TrackDto dto = null;
			dto = DaoUtil.track(content);
			if(null == dto)
				return null;
			dto.setPlatform(platform_type);
			String url = "";
			if(StringUtils.isNotBlank(dto.getUrl())){
				url = dto.getUrl();
				String searchWord = get_search_word(dto.getUrl());
				if(StringUtils.isNotBlank(searchWord)){
					TrackSearchWord tsw = new TrackSearchWord(dto.getPlatform(), dto.getBusiLine(),
							searchWord);
					final String json_searchWord = JsonUtil.toJson(tsw);
					try {
						threadPool.submit(new Runnable() {
							public void run() {
								RedisUtil.cache.lpushString(Constant.SEARCH_WORD_TRACK, json_searchWord);
							}
						});
					} catch (Exception e) {
						log.error("",e);
					}
				}
			}
			
			//过滤掉sou.58.com，只有pc端有
			if(url.contains("sou.58.com") || url.contains("cast.58.com") || "sou".equals(dto.getPageType()) ){
				return null;
			}
				
			return dto;
		}else
			return null;
	} 


	/* (non-Javadoc)
	 * @see com._58city.spark.app.mr.MrKafkaInterface#mapPair(java.util.List)
	 */
	public List<JavaPairDStream<String, TrackValue>> mapPair( List<JavaPairDStream<String, String>> kafkaStreams) {
		List<JavaPairDStream<String, TrackValue>> list =  new ArrayList<JavaPairDStream<String,TrackValue>>();
		for(JavaPairDStream<String, String> stream : kafkaStreams){
			JavaPairDStream<String, TrackValue> mapStream = stream.mapToPair(new PairFunction<Tuple2<String,String>, String, TrackValue>() {
						private static final long serialVersionUID = 3742490077169282334L;
						public Tuple2<String, TrackValue> call(Tuple2<String, String> event) throws Exception {
							try {
								String content = event._2;
								TrackDto dto = toDto(content);
									
								if(null != dto){
									//根据平台、业务线组合key，用于reduceByKey
									String key = SparkUtil.getMapKeyNotTime(dto, keys);
									if(null != key){
										long pv = 1;//每条日志算一个pv
										String cookie = dto.getCookie();
										Set<String> cookies = new HashSet<String>();
										cookies.add(cookie);
										String pageType = dto.getPageType();
										long listPv = 0;
										long vpPv = 0;
										//根据页面类型判断vpPv，listPv
										if(StringUtils.isNotBlank(pageType)){
											if(pageType.contains(TrackDto.PAGE_TYPE_DETAIL)) vpPv = 1;
											else if(pageType.contains(TrackDto.PAGE_TYPE_LIST)) listPv = 1;
										}
										TrackValue value = new TrackValue(pv, listPv, vpPv, cookies);
										return new Tuple2<String, TrackValue>(key, value);
									}
								}
							} catch (Exception e) {
								log.error("Track map ERROR!", e);
							}
							return new Tuple2<String, TrackValue>(SparkUtil.ERROR, new TrackValue());
						}
			});
			list.add(mapStream);
		}
		return list;
	}

	/* (non-Javadoc)
	 * @see com._58city.spark.app.mr.MrKafkaInterface#reducePair(org.apache.spark.streaming.api.java.JavaPairDStream)
	 */
	public JavaPairDStream<String, TrackValue> reducePair(JavaPairDStream<String, TrackValue> mapStream) {
		
		JavaPairDStream<String, TrackValue> reducePair = mapStream.reduceByKey(new Function2<TrackValue, TrackValue, TrackValue>() {
					private static final long serialVersionUID = 6264099295078802382L;
					
					//reduce加和
					public TrackValue call(TrackValue v1, TrackValue v2)
							throws Exception {
						try {
							long pv = v1.getPv() + v2.getPv();
							long listPv = v1.getListPv() + v2.getListPv();
							long vpPv = v1.getVpPv() + v2.getVpPv();
							Set<String> cookies = new HashSet<String>();
							cookies.addAll(v1.getCookies());
							cookies.addAll(v2.getCookies());
							return new TrackValue(pv, listPv, vpPv, cookies);
						} catch (Exception e) {
							log.error("Track reduce ERROR!",e);
						}
						return new TrackValue();
					}
				});
		return reducePair;
	}

	/* (non-Javadoc)
	 * @see com._58city.spark.app.mr.MrKafkaInterface#foreachRDD(org.apache.spark.streaming.api.java.JavaPairDStream)
	 */
//	public void foreachRDD(JavaPairDStream<String, TrackValue> reduceStream) {
//		reduceStream.foreachRDD(new Function<JavaPairRDD<String,TrackValue>, Void>() {
//			private static final long serialVersionUID = -8199089522425453338L;
//
//			public Void call(JavaPairRDD<String, TrackValue> v)
//					throws Exception {
//				long now = new Date().getTime();
//			    final long current_time = SparkUtil.makeFormateDate(now, batchInterval);
//				
//				v.foreach(new VoidFunction<Tuple2<String,TrackValue>>() {
//					private static final long serialVersionUID = 5929818265457879114L;
//
//					public void call(Tuple2<String, TrackValue> tuple) throws Exception {
//						try {
//							if(null != tuple){
//								String keyStr = tuple._1;
//								if(!SparkUtil.ERROR.equals(keyStr)){
//									TrackValue value = tuple._2;
//									String[] key_values = keyStr.split(",");
//									
//									long pv = value.getPv();
//									long listPv = value.getListPv();
//									long vpPv = value.getVpPv();
//									int uv = value.getCookies().size();	
//									log.info("Track ========== key : " + keyStr 
//												+ ",time : " + current_time
//												+ ",pv : " + pv + ",listPv : " + listPv
//												+ ",vpPv : " + vpPv
//												+ ",uv : " + uv);
//									TrackModel model = new TrackModel();
//									model.setTime(current_time);
//									TransferDimensionUtil.dimension(model, keys, Integer.parseInt(key_values[0]),
//											key_values[1]);
//									model.setPv(pv);
//									model.setListpv(listPv);
//									model.setVppv(vpPv);
//									model.setUv(uv);
//									try {
//										RedisUtil.cache.lpushString(Constant.TRACK_REDIS_QUEUE_KEY,
//												JsonUtil.toJson(model));
//									} catch (Exception e) {
//										log.error("track redis insert error!");
//									}
//									}
//								}
//						} catch (Exception e) {
//							log.error("Track foreachRDD ERROR!",e);
//						}
//					}
//				});
//				return null;
//			}
//		});
//		
//	}
	
	public void foreachRDD(JavaPairDStream<String, TrackValue> reduceStream) {
		reduceStream.foreachRDD(new Function<JavaPairRDD<String,TrackValue>, Void>() {
			private static final long serialVersionUID = -8199089522425453338L;

			public Void call(JavaPairRDD<String, TrackValue> v)throws Exception {
				long now = new Date().getTime();//当前时间戳
			    final long current_time = SparkUtil.makeFormateDate(now, batchInterval); //取batchInterval的整数倍
				
				v.foreachPartition(new VoidFunction<Iterator<Tuple2<String,TrackValue>>>() {
					private static final long serialVersionUID = 3040955589442973044L;

					public void call(Iterator<Tuple2<String, TrackValue>> iter)
							throws Exception {
						if(null != iter){
							final List<TrackModel> list = new ArrayList<TrackModel>();
						    while(iter.hasNext()){
								Tuple2<String, TrackValue> tuple = iter.next();
								String keyStr = tuple._1;
								if(!SparkUtil.ERROR.equals(keyStr)){
									TrackValue value = tuple._2;
									String[] key_values = keyStr.split(",");
									
									long pv = value.getPv();
									long listPv = value.getListPv();
									long vpPv = value.getVpPv();
								    final Set<String> cookies = value.getCookies();
								    int uv = cookies.size();
								    //APP的cookie由启动日志算，此处不存入redis
									if(uv > 0 && platform_type !=2){
										threadPool .submit(new Runnable() {
												public void run() {
													String date = DateFormatUtils.format(new Date(), "yyyyMMdd");
													String cookieKey = String.format(cookie_redis_key, date);
													RedisUtil.sCookieAdd(cookieKey, cookies);
												}
										});
									}
//									log.info("Track ========== key : " + keyStr 
//												+ ",time : " + current_time
//												+ ",pv : " + pv + ",listPv : " + listPv
//												+ ",vpPv : " + vpPv
//												+ ",uv : " + uv);
									TrackModel model = new TrackModel();
									model.setTime(current_time);
									TransferDimensionUtil.dimension(model, keys, Integer.parseInt(key_values[0]),
											key_values[1]);
									model.setPv(pv);
									model.setListpv(listPv);
									model.setVppv(vpPv);
									model.setUv(uv);
									list.add(model);
								}
							}
						    
						    if(list.size() > 0){
						    	log.info("Track {} {}: output {} recodes ." ,platform_type ,current_time,list.size());
						    	//结果写入redis
						    	threadPool.submit(new Runnable() {
									public void run() {
	                                  RedisUtil.send(Constant.TRACK_REDIS_QUEUE_KEY, list);										
									}
							});
						    }
						}
					}
					
				});
				return null;
			}
		});
		
	}

}
